package com.bujunjie.study.realtime.dwd.db.split.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.fastjson.parser.Feature;
import com.bujunjie.study.realtime.common.base.BaseApp;
import com.bujunjie.study.realtime.common.constant.FlinkConstant;
import com.bujunjie.study.realtime.common.enums.FlinkOpEnum;
import com.bujunjie.study.realtime.common.util.FlinkSinkUtil;
import com.bujunjie.study.realtime.common.util.FlinkSourceUtil;
import com.bujunjie.study.realtime.common.model.vo.DwdTableVO;
import com.bujunjie.study.realtime.dim.model.vo.Mysql2FlinkCdcVO;
import com.bujunjie.study.realtime.dim.model.vo.Mysql2MaxwellVO;
import com.bujunjie.study.realtime.dim.model.vo.ProductVO;
import com.bujunjie.study.realtime.dwd.db.split.function.DwdTableProcessFunction;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Map;

/**
 * <p>处理逻辑比较简单的事实表动态分流处理</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/15 10:00
 */
@Slf4j
public class DwdBaseDb extends BaseApp {

    public static void main(String[] args) throws Exception {
        new DwdBaseDb().start(10019,
                4,
                "dwd_base_db",
                FlinkConstant.TOPIC_DB
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaDS) {
        SingleOutputStreamOperator<Mysql2MaxwellVO<ProductVO>> dataDS = kafkaDS.process(new ProcessFunction<String, Mysql2MaxwellVO<ProductVO>>() {
            @Override
            public void processElement(String value, ProcessFunction<String, Mysql2MaxwellVO<ProductVO>>.Context ctx, Collector<Mysql2MaxwellVO<ProductVO>> out) throws Exception {
                try {
                    Mysql2MaxwellVO maxwellVO = JSON.parseObject(value, new TypeReference<Mysql2MaxwellVO<ProductVO>>() {
                    }, Feature.SupportNonPublicField, Feature.AllowUnQuotedFieldNames);
                    String type = maxwellVO.getType();
                    // maxwell 启动时，会发送bootstrap-insert，这里过滤掉
                    if (type.startsWith("bootstrap-")) {
                        out.collect(maxwellVO);
                    }
                } catch (Exception e) {
                    log.error("value=【{}】 转换异常", value);
                }
            }
        });

        // 创建MysqlSource对象
        MySqlSource<String> mySqlSource = FlinkSourceUtil.getMySqlSource("gmall2024_config", "table_process_dwd");
        //  转换成
        DataStreamSource<String> mysqlStrDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-source");
        SingleOutputStreamOperator<DwdTableVO> dtDS = mysqlStrDS.map(new MapFunction<String, DwdTableVO>() {
            @Override
            public DwdTableVO map(String value) throws Exception {
                Mysql2FlinkCdcVO<DwdTableVO> flinkCdcVO = JSON.parseObject(value, new TypeReference<Mysql2FlinkCdcVO<DwdTableVO>>() {
                        },
                        Feature.SupportNonPublicField, Feature.AllowUnQuotedFieldNames);
                String op = flinkCdcVO.getOp();
                DwdTableVO dwdTableVO = null;
                if (FlinkOpEnum.D.getOp().equals(op)) {
                    dwdTableVO = flinkCdcVO.getBefore();
                } else {
                    dwdTableVO = flinkCdcVO.getAfter();
                }
                dwdTableVO.setOp(op);
                return dwdTableVO;
            }
        });
        //  对配置流进行广播处理
        MapStateDescriptor<String, DwdTableVO> mapStateDescriptor = new MapStateDescriptor<>("dwdTableDesc", String.class, DwdTableVO.class);
        BroadcastStream<DwdTableVO> broadcastDS = dtDS.broadcast(mapStateDescriptor);
        // 关联主流和广播流
        BroadcastConnectedStream<Mysql2MaxwellVO<ProductVO>, DwdTableVO> connectDS = dataDS.connect(broadcastDS);
        SingleOutputStreamOperator<Tuple2<Map<String, Object>, DwdTableVO>> splitDS = connectDS.process(new DwdTableProcessFunction(mapStateDescriptor));
        splitDS.sinkTo(FlinkSinkUtil.getKafkaSink());
    }
}
